package com.test.reactor.ws;
import com.alibaba.fastjson.JSONObject;
import com.test.reactor.utils.ActorUtils;

import com.test.reactor.utils.SnowFlakeIdGenerator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Optional;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
 
/**
 * MessageHandler class
 *
 * @author zhaoyj
 * @date 2019/3/12
 */
@Component
@Slf4j
public class MessageHandler implements WebSocketHandler {
 
    @Autowired
    private SocketSessionRegistry sessionRegistry;
    @Autowired
    private ScheduledThreadPoolExecutor executor;
   
    @Autowired
    Executor myServiceExecuto;
    @Autowired
    private ActorUtils actorUtils;
    
    int loginInterval = 0;
    int pingInterval = 0;
    
    @Override
    public Mono<Void> handle(WebSocketSession session) {
 
        return session.receive().doOnSubscribe(s -> {
            log.info("发起连接:{}",s);
            /**
             * 你有10秒时间登陆，不登陆就关掉连接;并且不给任何错误信息
             */
            if(loginInterval != 0){
                executor.schedule(() -> sessionRegistry.checkAndRemove(session),loginInterval,TimeUnit.SECONDS);
            }
            if(pingInterval != 0){
                executor.schedule(() ->  session.send(Flux.just(session.pingMessage(DataBufferFactory::allocateBuffer))).toProcessor(),pingInterval, TimeUnit.SECONDS);
            }
        }).doOnTerminate(() -> {
            sessionRegistry.unregisterSession(session);
            myServiceExecuto.execute(() -> Optional.ofNullable(session).ifPresent(s -> s.close()));
            log.info("doOnTerminate");
        }).doOnComplete(() -> {
            log.info("doOnComplete");
        }).doOnCancel(() -> {
            log.info("doOnCancel");
        }).doOnNext(new AcceptMessge(session)).doOnError(e -> {
            e.printStackTrace();
            log.error("doOnError");
        }).doOnRequest(r -> {
            log.info("doOnRequest");
        }).then();
    }
    class AcceptMessge implements Consumer<WebSocketMessage>{
    	private WebSocketSession session;
    	public AcceptMessge(WebSocketSession session) {
    		this.session = session;
    	}
    	@Override
		public void accept(WebSocketMessage message) {
	            if(message.getType().equals(WebSocketMessage.Type.BINARY)){
	                log.info("收到二进制消息");
	            }else if(message.getType().equals(WebSocketMessage.Type.TEXT)){
	                String content = message.getPayloadAsText();
	                log.info("收到文本消息:{}",content);
	                if (content.startsWith("name:")) {
	                	login(content.replace("name:", ""), session);
	                } else {
	                	try{
	                		MyMessage myMessage = JSONObject.parseObject(content, MyMessage.class);
							myMessage.setMsgId(SnowFlakeIdGenerator.nextIdSingle());
	                		actorUtils.sendMessage(myMessage);
	                	}catch (Exception e){
	                		JSONObject obj = new JSONObject();
	                		obj.put("content","无法理解你发过来的消息内容，不予处理:"+content);
	                		session.send(Flux.just(session.textMessage(obj.toJSONString()))).then().toProcessor();
	                		log.error("解析消息内容出错", e);
	                		return;
	                	}
	                }
	            }else if(message.getType().equals(WebSocketMessage.Type.PING)){
	                session.send(Flux.just(session.pongMessage(s -> s.wrap(new byte[256]))));
	                log.info("收到ping消息");
	            }else if(message.getType().equals(WebSocketMessage.Type.PONG)){
	                log.info("收到pong消息");
	                if(pingInterval != 0){
	                    executor.schedule(() ->  session.send(Flux.just(session.pingMessage(DataBufferFactory::allocateBuffer))).toProcessor(),pingInterval, TimeUnit.SECONDS);
	                }
	            }
		}
    	private void login(String user, WebSocketSession session2) {
    		try {
    			actorUtils.creatActor(user, session2);
    			sessionRegistry.registerSession(session2, user);
			} catch (akka.actor.InvalidActorNameException e) {
				session2.send(Flux.just(session2.textMessage(user + "-- 已 登录"))).then().toProcessor();
				session2.close().toProcessor();
			}
    	}
    }
}